/*
Copyright 2024 - 2025 Zen HuiFer

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package main

import (
	"context"
	"flag"
	"fmt"
	influxdb2 "github.com/influxdata/influxdb-client-go/v2"
	"github.com/prometheus/client_golang/prometheus/promhttp"
	amqp "github.com/rabbitmq/amqp091-go"
	"go.mongodb.org/mongo-driver/mongo"
	"go.mongodb.org/mongo-driver/mongo/options"
	"go.uber.org/zap"
	"gopkg.in/yaml.v3"
	"log"
	"net/http"
	_ "net/http/pprof"
	"net/url"
	"os"
	"strconv"
)

var globalConfig ServerConfig

func main() {

	var configPath string
	flag.StringVar(&configPath, "config", "app-local-pre_handler.yml", "Path to the config file")
	flag.Parse()

	yfile, err := os.ReadFile(configPath)
	if err != nil {
		zap.S().Fatalf("error: %+v", err)
	}

	err = yaml.Unmarshal(yfile, &globalConfig)
	if err != nil {
		zap.S().Fatalf("error: %+v", err)
	}
	InitLog()

	InitGlobalRedisClient(globalConfig.RedisConfig)
	InitInfluxDbClient(globalConfig.InfluxConfig)
	//InitRabbitCon(globalConfig.MQConfig)
	err = ConnectToRMQ()
	if err != nil {
		log.Fatalf("Failed to connect to RabbitMQ: %s", err)
	}
	zap.S().Infof("消息队列类型 %s", globalConfig.NodeInfo.Type)

	CreateRabbitQueue("calc_queue")
	CreateRabbitQueue("waring_handler")
	CreateRabbitQueue("waring_notice")
	CreateRabbitQueue("transmit_handler")
	CreateRabbitQueue("waring_delay_handler")
	CreateRabbitQueue("pre_handler")
	CreateRabbitQueue("pre_tcp_handler")
	CreateRabbitQueue("pre_http_handler")
	CreateRabbitQueue("pre_ws_handler")
	CreateRabbitQueue("pre_coap_handler")
	initMongo()
	failOnError(err, "Failed to open a channel")
	go startHttp()
	cus := NewConsumer("", genUrl(globalConfig.MQConfig), "", "", "")
	err = cus.Connect()
	if err != nil {
		log.Fatalf("Failed to connect to RabbitMQ: %s", err)
	}

	if globalConfig.NodeInfo.Type == "pre_handler" {
		deliveries, err := cus.AnnounceQueue("pre_handler", "")
		if err != nil {
			log.Fatalf("Failed to connect to RabbitMQ: %s", err)
		}
		cus.Handle(deliveries, HandlerDataStorage, 10, "pre_handler", "")
	}
	if globalConfig.NodeInfo.Type == "waring_handler" {
		waringHandler, err := cus.AnnounceQueue("waring_handler", "")
		if err != nil {
			log.Fatalf("Failed to connect to RabbitMQ: %s", err)
		}
		cus.Handle(waringHandler, HandlerWaring, 1, "waring_handler", "")
	}
	if globalConfig.NodeInfo.Type == "calc_queue" {
		calcQueue, err := cus.AnnounceQueue("calc_queue", "")
		if err != nil {
			log.Fatalf("Failed to connect to RabbitMQ: %s", err)
		}
		cus.Handle(calcQueue, HandlerCalc, 1, "calc_queue", "")
	}
	if globalConfig.NodeInfo.Type == "waring_delay_handler" {
		waringDelayHandler, err := cus.AnnounceQueue("waring_delay_handler", "")
		if err != nil {
			log.Fatalf("Failed to connect to RabbitMQ: %s", err)
		}
		cus.Handle(waringDelayHandler, HandlerWaringDelay, 1, "waring_delay_handler", "")
	}

	if globalConfig.NodeInfo.Type == "transmit_handler" {
		transmitHandler, err := cus.AnnounceQueue("transmit_handler", "")
		if err != nil {
			log.Fatalf("Failed to connect to RabbitMQ: %s", err)
		}
		cus.Handle(transmitHandler, HandlerTransmit, 1, "transmit_handler", "")
	}
	if globalConfig.NodeInfo.Type == "waring_notice" {
		waringNotice, err := cus.AnnounceQueue("waring_notice", "")
		if err != nil {
			log.Fatalf("Failed to connect to RabbitMQ: %s", err)
		}
		cus.Handle(waringNotice, HandlerNotice, 1, "waring_notice", "")
	}



	// 协议层处理
	if globalConfig.NodeInfo.Type == "pre_tcp_handler" {
		preTcpHandler, err := cus.AnnounceQueue("pre_tcp_handler", "")
		if err != nil {
			log.Fatalf("Failed to connect to RabbitMQ: %s", err)
		}
		cus.Handle(preTcpHandler, HandlerTcpDataStorage, 1, "pre_tcp_handler", "")
	}

	if globalConfig.NodeInfo.Type == "pre_http_handler" {
		preHttpHandler, err := cus.AnnounceQueue("pre_http_handler", "")
		if err != nil {
			log.Fatalf("Failed to connect to RabbitMQ: %s", err)
		}
		cus.Handle(preHttpHandler, HandlerHttpDataStorage, 1, "pre_http_handler", "")
	}
	if globalConfig.NodeInfo.Type == "pre_ws_handler" {
		preWsHandler, err := cus.AnnounceQueue("pre_ws_handler", "")
		if err != nil {
			log.Fatalf("Failed to connect to RabbitMQ: %s", err)
		}
		cus.Handle(preWsHandler, HandlerWsDataStorage, 1, "pre_ws_handler", "")
	}
	if globalConfig.NodeInfo.Type == "pre_coap_handler" {
		preCCoapHandler, err := cus.AnnounceQueue("pre_coap_handler", "")
		if err != nil {
			log.Fatalf("Failed to connect to RabbitMQ: %s", err)
		}
		cus.Handle(preCCoapHandler, HandlerCoapDataStorage, 1, "pre_coap_handler", "")
	}

}

func startHttp() {
	http.HandleFunc("/beat", HttpBeat)
	http.Handle("/metrics", promhttp.Handler())

	if err := http.ListenAndServe(":"+strconv.Itoa(globalConfig.NodeInfo.Port), nil); err != nil {
		zap.S().Fatalf("Failed to start server: %s", err)
	}
	zap.S().Infof("Server started Port %d", globalConfig.NodeInfo.Port)
}

func HttpBeat(w http.ResponseWriter, r *http.Request) {
	// 检查请求方法
	if r.Method != http.MethodGet {
		http.Error(w, "Only GET method is allowed", http.StatusMethodNotAllowed)
		return
	}
	w.Header().Set("Access-Control-Allow-Origin", "*")
	// 允许的请求方法
	w.Header().Set("Access-Control-Allow-Methods", "GET, POST, PUT, DELETE")
	// 允许的请求头部
	w.Header().Set("Access-Control-Allow-Headers", "Origin, X-Requested-With, Content-Type, Accept, Authorization")

	// 非简单请求时，浏览器会先发送一个预检请求(OPTIONS)，这里处理预检请求
	if r.Method == http.MethodOptions {
		w.WriteHeader(http.StatusNoContent) // 200 OK 也可以
		return
	}
	// 向客户端发送响应消息
	_, _ = fmt.Fprintf(w, "ok")
}

func failOnError(err error, msg string) {
	if err != nil {
		zap.S().Panicf("%s: %s", msg, err)
	}
}

var GlobalInfluxDbClient influxdb2.Client

// InitInfluxDbClient 函数用于初始化InfluxDB客户端
//
// 参数：
// config InfluxConfig - InfluxDB配置信息
//
// 返回值：
// 无
func InitInfluxDbClient(config InfluxConfig) {
	GlobalInfluxDbClient = influxdb2.NewClient(fmt.Sprintf("http://%s:%d", config.Host, config.Port), config.Token)

}

// PushToQueue 将消息推送到RabbitMQ队列中
//
// 参数：
// queueName string - 目标队列名称
// body []byte - 消息体
//
// 返回值：
// 无
func PushToQueue(queueName string, body []byte) {

	zap.S().Debugf("开始推送消息到队列 %s msg %s", queueName, body)

	err := chann.PublishWithContext(context.Background(), "", queueName, // routing key
		false, // mandatory
		false, // immediate
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        body,
		})
	failOnError(err, "Failed to open a channel")

}

var GRabbitMq *amqp.Connection

// InitRabbitCon 初始化RabbitMQ连接
func InitRabbitCon(config MQConfig) {
	zap.S().Infof("开始处理rabbitmq")
	conn, err := amqp.Dial(genUrl(config))
	if err != nil {
		zap.S().Fatalf("Failed to connect to RabbitMQ  %+v", err)
	}

	GRabbitMq = conn
	zap.S().Infof("Connected to RabbitMQ")

}

// genUrl 生成RabbitMQ连接字符串
//
// 返回值为一个字符串，格式为amqp://用户名:密码@主机名:端口号/
//
// 参数说明：
//
//	无
//
// 返回值说明：
//
//	返回RabbitMQ的连接字符串
func genUrl(config MQConfig) string {
	connStr := fmt.Sprintf("amqp://%s:%s@%s:%d/", config.Username, config.Password, config.Host, config.Port)
	return connStr
}

// CreateRabbitQueue 创建一个RabbitMQ队列
//
// 参数：
// queueName string - 队列名称
//
// 返回值：
// 无
func CreateRabbitQueue(queueName string) {

	_, err := chann.QueueDeclare(queueName, // name
		true,  // durable
		false, // delete when unused
		false, // exclusive
		false, // no-wait
		nil,   // arguments
	)
	if err != nil {
		zap.S().Fatalf("创建queue异常 %s", queueName)
	}
}

var GMongoClient *mongo.Client

// initMongo 函数用于初始化 MongoDB 连接
func initMongo() {

	connStr := fmt.Sprintf("mongodb://%s:%s@%s:%d", url.QueryEscape(globalConfig.MongoConfig.Username), url.QueryEscape(globalConfig.MongoConfig.Password), globalConfig.MongoConfig.Host, globalConfig.MongoConfig.Port)
	client, err := mongo.Connect(context.TODO(), options.Client().ApplyURI(connStr))
	if err != nil {
		log.Fatal(err)
	}

	// 检查连接
	err = client.Ping(context.TODO(), nil)
	if err != nil {
		log.Fatal(err)
	}
	GMongoClient = client

	zap.S().Infof("Connected to MongoDB!")

}
